package com.sam.kafka.producer;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.common.serialization.StringSerializer;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * @author: wm.xue
 * @date: 2021/11/3 14:27
 * @description:
 */
public class _00_MyProducer {

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        // KafkaProducer配置
        Map<String, Object> configs = new HashMap<>();
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.100.17.117:9092");
        configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        configs.put(ProducerConfig.RETRIES_CONFIG, 2);
        configs.put(ProducerConfig.ACKS_CONFIG, "1");
        configs.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1);
        configs.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, 4);

        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(configs);

        // 用户自定义消息头
        List<Header> headers = new ArrayList<>(16);
        headers.add(new RecordHeader("myHeader", "_00_MyProducer".getBytes(StandardCharsets.UTF_8)));


        ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(
                "tp_demo_00",
                0,
                String.valueOf(0),
                "hello kafka 0",
                headers
        );


        // 同步消息发送
        Future<RecordMetadata> futureSend = producer.send(producerRecord);

        RecordMetadata recordMetadata = futureSend.get();
        System.err.println("消息的主题：" + recordMetadata.topic());
        System.err.println("消息的分区：" + recordMetadata.partition());
        System.err.println("消息的偏移量：" + recordMetadata.offset());



//        producer.send(producerRecord, new Callback() {
//            @Override
//            public void onCompletion(RecordMetadata metadata, Exception exception) {
//                if (null == exception) {
//                    System.err.println("消息的主题：" + metadata.topic());
//                    System.err.println("消息的分区：" + metadata.partition());
//                    System.err.println("消息的偏移量：" + metadata.offset());
//                }else {
//                    System.out.println("消息异常,异常原因:" + exception.getMessage());
//                }
//            }
//        });

//        producer.send(producerRecord,(RecordMetadata metadata, Exception exception) -> {
//            if (null == exception) {
//                System.err.println("消息的主题：" + metadata.topic());
//                System.err.println("消息的分区：" + metadata.partition());
//                System.err.println("消息的偏移量：" + metadata.offset());
//            }else {
//                System.out.println("消息异常,异常原因:" + exception.getMessage());
//            }
//        });

        producer.close();
    }

}
